package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_Exam_issue extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_Exam_issue().init(2004,"Dwd_Exam_issue");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_Exam_issue");
        Table tableResult = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['exam_id'] exam_id, " +
                "data['paper_id'] paper_id, " +
                "data['question_id'] question_id, " +
                "data['user_id'] user_id, " +
                "data['answer'] answer, " +
                "data['is_correct'] is_correct, " +
                "data['score'] score, " +
                "data['create_time'] create_time, " +
                "data['update_time'] update_time, " +
                "data['deleted'] deleted, " +
                "ts " +
                "from ods_db " +
                "where `table`='test_exam_question' " +
                "and `type`='insert' ");
        tEnv.executeSql("create table dwd_exam_issue(" +
                "id string, " +
                "exam_id string, " +
                "paper_id string, " +
                "question_id string, " +
                "user_id string,  " +
                "answer string, " +
                "is_correct string, " +
                "score string, " +
                "create_time string, " +
                "update_time string, " +
                "deleted string, " +
                "ts string " +
                ")"+ SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_EXAM_ISSUE));
        tableResult.executeInsert("dwd_exam_issue");
    }
}
